package com.auto.storm;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.RandomAccessFile;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.Map;

import lombok.extern.java.Log;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

/** 
* @ClassName: SocketSpout 
* @Description: 数据处理
* @author hcw 
* @date 2016年12月26日 下午7:09:25 
*  
*/
@Log
public class SocketSpout extends BaseRichSpout{
	
    //socket IP 和端口
    static String url ="10.122.1.133"; 
    static int port =1234;
    /**
	 * 配置socket 和接收传过来的数据大小
	 */
	static Socket sock=null;  
    static BufferedReader in=null;
    String str=null;  
    private static final long serialVersionUID = 1L;  
    private SpoutOutputCollector _collector;  
    private BufferedReader br;  
    private String dataFile;  
    private BufferedWriter bw2;  
    RandomAccessFile randomFile;  
    int cnt=0;  
     	/**
     	 * 定义如何读取spout文件  
     	 */
    @Override
	public void open(Map conf, TopologyContext context,
			SpoutOutputCollector collector) {
		  _collector = collector;  
		  if(sock==null){
 	    	 try {
				sock=new Socket(url,port);
			} catch (UnknownHostException e) {
				e.printStackTrace();
			} catch (IOException e) {
				e.printStackTrace();
			}
 	     }
	}
    /**
     * 把流数据传给blot处理
     */
    @Override
	public void nextTuple() {
    	  try {
    	     if(sock==null){
    	    	 sock=new Socket(url,port);
    	     }
    	     while(true){
    	    	 byte[] b3= new byte[200];
    	    	 if(sock!=null ){
    	    		 try{
    	    			 sock.getInputStream().read(b3);
    	    		 }catch ( Exception e){
    	    			 System.out.println("connection reset, reconnecting ...");
    	    			 sock.close();
    	    			 Thread.sleep(100);
    	    			 sock=new Socket(url,port);;
    	    		 }

    	    	 }else{
    	    		 sock=new Socket(url,port);;
    	    		 break ;
    	    	 }
    	    	 System.out.println("服务端首次收到的数据为："+new String(b3));
    	    	 _collector.emit(new Values(new String(b3)));
    	     } 
    	  	}catch (IOException e) {
    	      e.printStackTrace();
    	     } catch (Exception e) {
    	      e.printStackTrace();
    	     }

	}
    @Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
    	  declarer.declare(new Fields("line"));  
	}
}
